package com.eshipenet.shipnet.equipmentdatahub

import java.util.Properties

import org.apache.flink.api.java.tuple.Tuple2
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import play.api.libs.json.{JsValue, Json}

import scala.language.postfixOps
import scala.util.Try

object EdgeChannelDataStreamProvider {

  def createEdgeChannelDataStream(env: StreamExecutionEnvironment): DataStream[JsValue] = {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "kafka-server:9092")
    properties.setProperty("group.id", "equipmentdatahub.EdgeChannelHub")
    val der      = new EdgeChannelDataKafkaDeserializer()
    val consumer = new FlinkKafkaConsumer[RawKafkaMessage]("EdgeChannelData2", der, properties)
    consumer.setStartFromLatest()
    val stream = env
      .addSource(consumer)
      .map(msg => Try(Json.parse(msg.value)))
      .filter(_.isSuccess)
      .map(_.get)
    stream
  }

}
